package com.shujia.sql

import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

import scala.language.postfixOps

object Demo03DSL {
  def main(args: Array[String]): Unit = {
    /**
     * SQL中常见的操作：
     * select
     * from tb
     * where 过滤
     * groupBy 分组，分组后一般会使用聚合函数（sum、max、min、avg、count）进行统计
     * having 聚合之后进行过滤
     * order by 排序
     * limit 限制查询的数据条数
     * 关联：多表之间横向的拼接
     * 内连接：join（inner join）
     * 外连接：left outer join （left join）、right outer join （right join）、full outer join（full join）
     * 联接：
     * union：去重
     * union all：不去重
     *
     * 常见的函数：
     * 聚合函数：sum、max、min、avg、count
     * 数值函数：round、floor、cell、log、cos、sin等等
     * 字符串函数：concat、concat_ws、substring、split等等
     * 条件函数：if、case when、coalesce
     * 日期函数：转换字符串、日期、时间戳之间的相互转换
     * 行列转换：explode、lateral view、collect_set、collect_list
     * 窗口函数：
     * 排名：row_number、rank、dense_rank
     * 取值：lag、lead、first_value
     * 聚合函数：sum、max、min、avg、count
     * 其他
     *
     *
     */

    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo03DSL")
      .master("local")
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    val stuDF: DataFrame = spark.read.format("csv").option("sep", ",").schema("id String,name String,age Int,gender String,clazz String")
      .load("spark/data/students.txt")

    // 对多次使用的DF进行缓存
    stuDF.cache()

    /**
     * DSL：Domain Specific Language 特定领域语言
     * 实际上是DataFrame提供的一种API，只能适用于SparkSQL中的DF
     * 在DataFrame中的每一条数据其实是一个Row对象
     * Row对象：有结构的一行数据，即有列名，可以通过列名取列值
     *
     * Dataset[Row] 等于 DataFrame
     */

    // 1、select 取出需要的字段 以及可以结合函数对字段进行处理
    stuDF.select("id", "name", "clazz")
    //      .show() // 字符串表达式
    stuDF.select($"id", $"name", $"clazz")
    //      .show() // 列表达式

    // 2、where 过滤
    // 过滤文科一班的学生并且性别为男
    stuDF.where("clazz='文科一班' and gender = '男'")
    //      .show()
    stuDF.where($"clazz" === "文科一班" and $"gender" === "女")
    //      .show()

    // 还能使用filter进行过滤
    stuDF.filter("clazz='文科一班' and gender = '男'")
    //      .show()
    stuDF.filter($"clazz" === "文科一班" and $"gender" === "女")
    //      .show()
    stuDF.filter((row: Row) => {
      val clazz: String = row.getAs[String]("clazz")
      val gender: String = row.getAs[String]("gender")
      var flag: Boolean = false
      if (clazz == "文科一班" && gender == "男") {
        flag = true
      }
      flag
    })
    //      .show()

    // 3、group By 分组 聚合
    // 统计班级人数
    stuDF.groupBy($"clazz").agg(count("id") as "cnt")
    //      .show()
    // 统计性别人数
    stuDF.groupBy($"gender").agg(count("id") as "cnt")
    //      .show()

    // 4、having 聚合之后进行过滤
    // 统计班级人数 并过滤出人数大于90的班级
    stuDF.groupBy($"clazz").agg(count("id") as "cnt").where($"cnt" > 90)
    //      .show()

    // 5、order by 排序 以及 limit 限制数据条数
    // 统计班级人数 按照人数降序排序 并 返回 前5名
    stuDF
      .groupBy($"clazz")
      // 调用函数之后要记得取别名
      .agg(count("id") as "cnt")
      .orderBy($"cnt".desc)
      .limit(5)
    //      .show()

    /**
     * 6、关联
     */
    // 加载分数数据
    val scoDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id String,subject String,score Int")
      .load("spark/data/score.txt")

    // 内连接：inner join （join）

    // 当关联的字段同名时，不能直接 $"id" === $"id"，需要把关联字段构造一个序列Seq作为关联条件
    //    stuDF.join(scoDF, $"id" === $"id").show()
    stuDF.join(scoDF, List("id")).show()

    // 打印DF的Schema
    scoDF.printSchema()
    // 对DataFrame某个字段重命名 并 返回一个新的DF
    val newScoDF: DataFrame = scoDF.withColumnRenamed("id", "stu_id")
    newScoDF.printSchema()
    // 当关联的字段名不相同时，可以通过列表达式指定关联条件：$"id" === $"stu_id"
    stuDF.join(newScoDF, $"id" === $"stu_id").show() // 默认使用inner join方式进行关联

    /**
     * 外连接：left join、right join、full join
     * 默认是inner join，通过指定joinType改变关联方式
     * joinType：
     * inner, cross, outer
     * , full, full_outer
     * , left, left_outer
     * , right, right_outer
     * , left_semi, left_anti.
     */

    // left join
    stuDF.join(newScoDF, $"id" === $"stu_id", "left").show()
    println(stuDF.join(newScoDF, $"id" === $"stu_id", "left").count()) // 5986

    // right join
    stuDF.join(newScoDF, $"id" === $"stu_id", "right").show()
    println(stuDF.join(newScoDF, $"id" === $"stu_id", "right").count()) // 5991

    // full join
    stuDF.join(newScoDF, $"id" === $"stu_id", "full").show()
    println(stuDF.join(newScoDF, $"id" === $"stu_id", "full").count()) // 5992

    // 7、联接操作：union、union all
    val sampleDF01: DataFrame = stuDF.sample(withReplacement = false, fraction = 0.01, seed = 1)
    val sampleDF02: Dataset[Row] = stuDF.sample(withReplacement = false, fraction = 0.01, seed = 1)

    sampleDF01.union(sampleDF02).show() // 相当于是union all 不会去重
    sampleDF01.union(sampleDF02).distinct().show() // 如果需要去重则在后面加上distinct操作


    // 8、case when 基于不同的条件实现不同的逻辑 注意在DSL中没有if
    // 基于学生的班级 判断学生是 文科 还是 理科
    stuDF
      .select($"id", $"name", $"age", $"gender", $"clazz"
        , when(substring($"clazz", 1, 2) === "理科", "理科")
          .when(substring($"clazz", 1, 2) === "文科", "文科")
          .otherwise("文科") as "cls")
      .show()

    // 如果只是想要在原有数据的基础之上增加新的一列 则可以使用withColumn方法
    stuDF
      .withColumn("cls", when(substring($"clazz", 1, 2) === "理科", "理科")
        .when(substring($"clazz", 1, 2) === "文科", "文科")
        .otherwise("文科"))
      .show()

  }

}
